package com.yfbao.horizon.docker.trx;


import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.yfbao.horizon.docker.common.CacheConstants;
import com.yfbao.horizon.docker.common.UserTrxConstants;
import com.yfbao.horizon.docker.dao.TrxCheckDao;
import com.yfbao.horizon.docker.dao.TrxHistoryDao;
import com.yfbao.horizon.docker.dao.TrxTaskDao;
import com.yfbao.horizon.docker.entity.TrxCheck;
import com.yfbao.horizon.docker.entity.TrxHistory;
import com.yfbao.horizon.docker.entity.TrxTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.CacheEvict;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 分布式事务处理类
 */
@Slf4j
@Component
public class CtpUserDTPServcie {

    @Autowired
    private TrxCheckDao trxCheckDao;
    @Autowired
    private TrxTaskDao trxTaskDao;
    @Autowired
    private TrxHistoryDao trxHistoryDao;
//    @Autowired
//    private DistributedLockFactory lockFactory;
//    @Autowired
//    private RedisTemplate redisTemplate;

    private final static Integer maxRunningTaskNum = UserTrxConstants.TASK_RETRY_QUEUE_LENGTH;
    //资源锁
    private static final Object obj = new Object();
    //任务添加线程是否启动
    private static volatile boolean RUN_STATUS =false;


    //todo:补全线程名等信息
    private ScheduledExecutorService scheduledThreadPool= new ScheduledThreadPoolExecutor(1,new TrxTaskThreadFactory("trxTask-runner"));


    private static Queue<Long>  waitRetryQueue =  new LinkedBlockingDeque<Long>(maxRunningTaskNum*100);
    private static Set<Long> runningSet = Collections.synchronizedSet(new HashSet<>());



    /**
     * 幂等检查
     * @param trxId 事务id
     * @param type  操作类型
     * @return 未处理过 false,处理过 true
     */
    public boolean trxCheck(Long trxId, String type){
        TrxHistory trxHistory = getTrxHistory(trxId, type);
        if(trxHistory ==null){
            return false;
        }
        return true;
    }

    //获取指定类型的任务执行记录
    public TrxHistory getTrxHistory(Long trxId, String type){
        List<TrxHistory> trxHistoryBytrxId = getTrxHistoryBytrxId(trxId);
        if(trxHistoryBytrxId == null){
            return null;
        }
        for (TrxHistory trxHistory : trxHistoryBytrxId) {
            if(trxHistory.getType().equals(type)){
                return trxHistory;
            }
        }
        return null;
    }

    //获取指定类型的任务执行记录
    public TrxHistory getTrxHistory(Long trxId, String type,Boolean taskRes){
        List<TrxHistory> trxHistoryBytrxId = getTrxHistoryBytrxId(trxId);
        if(trxHistoryBytrxId == null){
            return null;
        }
        for (TrxHistory trxHistory : trxHistoryBytrxId) {
            if(trxHistory.getType().equals(type) && trxHistory.getTaskRes().equals(taskRes)){
                return trxHistory;
            }
        }
        return null;
    }


    public List<TrxHistory> getTrxHistoryBytrxId(Long trxId){
        if(trxId ==null){
            //如果没有trxId，默认为没有处理过的请求
            return null;
        }
        List<TrxHistory> trxHistoryByTrxId = getTrxHistoryByTrxId(trxId);
        if(trxHistoryByTrxId ==null || trxHistoryByTrxId.isEmpty()){
            return null;
        }
        return trxHistoryByTrxId;
    }


    @CacheEvict(value = CacheConstants.TRX_HISTORY, allEntries=true)
    public void batchDeleteHistory(List<Long> batchDelTrxIds){
//        Wrapper wrapper = new Wrapper(TrxHistory.class);
//        wrapper.and().andIn("trxId",batchDelTrxIds);
//        Map<String,Object> map =new HashMap<>();
//        map.put("IN_trxId",batchDelTrxIds);
//        trxHistoryDao.deleteByConditionsIgnoreTenant(map);
        QueryWrapper<TrxHistory> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().in(TrxHistory::getTrxId,batchDelTrxIds);
        trxHistoryDao.delete(queryWrapper);
    }

    @Cacheable(value = CacheConstants.TRX_HISTORY, key = "#trxId")
    public List<TrxHistory> getTrxHistoryByTrxId(Long trxId){
//        Wrapper wrapper = new Wrapper(TrxHistory.class);
//        wrapper.and().andEqualTo("trxId",trxId);
        QueryWrapper<TrxHistory> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(TrxHistory::getTrxId,trxId);
        return trxHistoryDao.selectList(queryWrapper);
    }

    /**
     * 保存至已检查信息表
     * @param trxId
     * @param className
     * @param methodName
     * @param args
     */
    public void saveTrxCheck(Long trxId,String className, String methodName,Boolean checkRes,Object... args){
        if(trxId ==null){
            log.error("missing trxId ，unsupported class:"+className+",method:"+methodName);
            return;
        }
        TrxCheck trxCheck = new TrxCheck();
        trxCheck.setTrxId(trxId);
        trxCheck.setServiceName("ctp-user");
        trxCheck.setClassName(className);
        trxCheck.setMethod(methodName);
        if(args !=null){
            trxCheck.setParamSize(args.length);
            String params = parseToString(args);
            trxCheck.setParamData(params);
        }
        if(checkRes !=null){
            trxCheck.setCheckRes(checkRes);
        }
//        trxCheckDao.create(trxCheck);
        saveTrxCheckToHistory(trxCheck);
    }

    @CacheEvict(value = CacheConstants.TRX_HISTORY,key = "#trxCheck.trxId")
    public void saveTrxCheckToHistory(TrxCheck trxCheck){
        boolean trxIdCheck = trxCheck(trxCheck.getTrxId(), UserTrxConstants.CHECK);
//        AssertUtil.isNotTrue(trxIdCheck, UserErrorCode.PARAMETER_ERROR);
        TrxHistory trxHistroy = new TrxHistory();
        trxHistroy.setTrxId(trxCheck.getTrxId());
        trxHistroy.setServiceName("ctp-user");
        trxHistroy.setClassName(trxCheck.getClassName());
        trxHistroy.setMethod(trxCheck.getMethod());
        trxHistroy.setType(UserTrxConstants.CHECK);
        trxHistroy.setTaskRes(true);
        trxHistoryDao.insert(trxHistroy);
    }


    @CacheEvict(value = CacheConstants.TRX_HISTORY,key = "#trxTask.trxId")
    public void saveTrxTaskToHistory(TrxTask trxTask){
        boolean trxIdCheck = trxCheck(trxTask.getTrxId(), UserTrxConstants.COMMIT);
//        AssertUtil.isNotTrue(trxIdCheck, UserErrorCode.PARAMETER_ERROR);
        TrxHistory trxHistroy = new TrxHistory();
        trxHistroy.setTrxId(trxTask.getTrxId());
        trxHistroy.setServiceName("ctp-user");
        trxHistroy.setClassName(trxTask.getClassName());
        trxHistroy.setMethod(trxTask.getMethod());
        trxHistroy.setType(UserTrxConstants.COMMIT);
        if(!trxTask.getTaskStatus().equals(UserTrxConstants.TASK_FINISHED)){
            trxHistroy.setTaskRes(false);
        }
        trxHistoryDao.insert(trxHistroy);
    }

    @CacheEvict(value = CacheConstants.TRX_HISTORY,key = "#trxId")
    public void updateTrxTaskToHistory(Long trxId ,Boolean taskRes){
        if(taskRes == null){
            return;
        }
        TrxHistory trxHistory = getTrxHistory(trxId, UserTrxConstants.COMMIT);
        trxHistory.setTaskRes(taskRes);
        trxHistoryDao.updateById(trxHistory);
    }


    private String parseToString(Object... args){
        if(args ==null){
            return "";
        }
        List<String> list = new ArrayList<>(args.length);
        for (Object arg : args) {
            String s = JSON.toJSONString(arg);
            list.add(s);
        }
        return JSON.toJSONString(list);
    }

    /**
     * 检查任务并保存任务，同时开始执行任务
     * @param trxId
     * @param className
     * @param methodName
     * @param args
     */
    public void checkAndSaveTrxTaskRun(Long trxId,String className, String methodName,Object... args){
        boolean checkFinish = checkTrxTaskFinished(trxId);
//        AssertUtil.isNotTrue(checkFinish,UserErrorCode.PARAMETER_ERROR);
        boolean checkTrxTaskExist = checkTrxTaskExist(trxId);
//        AssertUtil.isNotTrue(checkTrxTaskExist,UserErrorCode.PARAMETER_ERROR);
        saveTrxTaskAndRun(trxId,className,methodName,args);
    }

    //保证下面执行必须成功
    //如果多次失败，可以考虑从check包中获取转存至本任务表
    public boolean saveTrxTaskAndRun(Long trxId,String className, String methodName,Object... args){
        if(trxId ==null){
            //如果没有trxId，默认按照非分布式事务处理
            return true;
        }
        TrxTask trxTask = new TrxTask();
        trxTask.setTrxId(trxId);
        trxTask.setServiceName("ctp-user");
        trxTask.setClassName(className);
        trxTask.setMethod(methodName);
        if(args !=null){
            trxTask.setParamSize(args.length);
            //todo： 要做数据长度检查，避免数据过长，数据库保存不了
            String params = parseToString(args);
            if(params.length() > 10000){
                log.warn("trx:"+trxId+",param length over max:"+params.length());
            }
            trxTask.setParamData(params);
        }
        trxTask.setTaskStatus(UserTrxConstants.TASK_RUNNING);

        Integer i = reliableSaveTrxTask(trxTask, 0);
        if(i<0){
            //todo:保存任务失败。记录日志，后期考虑
            log.error("save trxTask error,class:"+className+",method:"+methodName+",size:"+trxTask.getParamSize()+", params:"+trxTask.getParamData());
        }
        saveTrxTaskToHistory(trxTask);
        //todo： 记录至 redis运行队列，默认10s自动过期
        return true;
    }

    /**
     * 检查任务是否已经写入任务表
     * @param trxId
     * @return 没有写入 false;  已经写入 true
     */
    public boolean checkTrxTaskFinished(Long trxId){
        return trxCheck(trxId, UserTrxConstants.COMMIT);
    }

    /**
     * 检查任务是否已经写入任务表
     * @param trxId
     * @return 没有写入 false;  已经写入 true
     */
    public boolean checkTrxTaskExist(Long trxId){
        TrxTask trxTaskByTrxId = getTrxTaskByTrxId(trxId);
        if(trxTaskByTrxId !=null){
            return true;
        }
        return false;
    }

    @Cacheable(value = CacheConstants.TRX_TASK, key = "#trxId")
    public TrxTask getTrxTaskByTrxId(Long trxId){
        QueryWrapper<TrxTask> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(TrxTask::getTrxId,trxId);
        return trxTaskDao.selectOne(queryWrapper);
//        Map<String,Object> map = new HashMap<>();
//        map.put("trxId",trxId);
//        return trxTaskDao.selectOneByConditionsIgnoreTenant(map);
    }

    /**
     * 可靠保存，内置10次重试
     * @param trxTask
     * @param retryTime
     * @return
     */
    public Integer reliableSaveTrxTask(TrxTask trxTask,int retryTime){
        retryTime ++;
        try{
            return trxTaskDao.insert(trxTask);
        }catch (Throwable t){
            if(retryTime<10){
                try {
                    Thread.sleep(10*retryTime);
                } catch (InterruptedException e) {
                  log.error("thread sleep error",e);
                }
                return reliableSaveTrxTask(trxTask,retryTime);
            }
            log.error("save task error",t);
        }
        return -1;
    }


    //正常完成任务后，更新事务表
    public void updateTrxTaskSuccess(Long trxId){
        try {
            updateTrxTaskToHistory(trxId,true);
            reliableTrxTaskDelete(trxId,0);
        }catch (Throwable t){
            log.error("updateTrxTaskSuccess error:"+trxId,t);
        }
    }


    public void updateTrxTaskStatus(Long trxId, Integer taskStatus){
        TrxTask trxTask = getTrxTaskByTrxId(trxId);
        trxTask.setTaskStatus(taskStatus);
        reliableUpdateTrxTask(trxTask,0);
    }

    public void reliableUpdateTrxTask(TrxTask task, int retryTime){
        reliableUpdateTrxTask(task,retryTime,true);
    }

    @CacheEvict(value = CacheConstants.TRX_TASK,key = "#task.trxId")
    public void reliableUpdateTrxTask(TrxTask task,int retryTime,Boolean isRetry){
        try {
            trxTaskDao.updateById(task);
        }catch (Throwable t){
            if(isRetry && retryTime<10){
                try {
                    Thread.sleep(10*retryTime);
                } catch (InterruptedException e) {
                    log.error("thread sleep error",e);
                }
                reliableUpdateTrxTask(task,retryTime+1);
                return;
            }
            log.error("update task error",t);
        }
    }
    @CacheEvict(value = CacheConstants.TRX_TASK,key = "#trxId")
    public boolean reliableTrxTaskDelete(Long trxId,Integer retryTime){
        retryTime ++;
        try{
            QueryWrapper<TrxTask> queryWrapper = new QueryWrapper<>();
//            Wrapper wrapper = new Wrapper(TrxTask.class);
            queryWrapper.lambda().eq(TrxTask::getTrxId,trxId);
            trxTaskDao.delete(queryWrapper);
//            Map<String,Object> map = new HashMap<>();
//            map.put("trxId",trxId);
//            trxTaskDao.delete(map);
            return true;
        }catch (Throwable t){
            if(retryTime<10){
                try {
                    Thread.sleep(10*retryTime);
                } catch (InterruptedException e) {
                    log.error("thread sleep error",e);
                }
                return reliableTrxTaskDelete(trxId,retryTime);
            }
            log.error("delete task error",t);
        }
        return false;
    }

    public void addRetryTask(Long trxId){
        waitRetryQueue.add(trxId);
        runRetry();
    }

    /**
     * 启动添加任务线程
     * 如果已经再运行中，不在重复启动
     */
    public void runRetry(){
        if(RUN_STATUS){
            return;
        }
        synchronized (obj){
            if(RUN_STATUS){
                return;
            }
            Thread thread = new Thread(new TrxTaskRunner());
            thread.start();
            RUN_STATUS =true;
        }
    }


    /**
     * 从待执行队列中获取任务放入执行队列执行任务
     */
    class TrxTaskRunner implements Runnable {
        @Override
        public void run() {
            try {
                while (true) {
                    if (!RUN_STATUS) {
                        //如果发现运行状态为false, 更新执行状态
                        synchronized (obj) {
                            if(!RUN_STATUS){
                                RUN_STATUS = true;
                            }
                        }
                    }
                    if (!waitRetryQueue.isEmpty()) {
                        Long poll = waitRetryQueue.poll();
                        if (poll != null) {
                            if (runningSet.size() < maxRunningTaskNum) {
                                runningSet.add(poll);
                                RetryTrxTaskRunner retryTrxTaskRunner = new RetryTrxTaskRunner(poll, 10, 0);
                                scheduledThreadPool.schedule(retryTrxTaskRunner, 5, TimeUnit.MILLISECONDS);
                            }else {
                                log.info("及时任务补偿队列已满..., trx:"+poll);
                            }
                        }
                    }
                    Thread.sleep(200); //固定睡眠，释放cpu
                    if (waitRetryQueue.isEmpty()) {
                        break;
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                synchronized (obj) {
                    //完成任务后更新状态
                    RUN_STATUS = false;
                }
            }
        }
    }

    /**
     * 分布式事务及时重试
     */
    class RetryTrxTaskRunner implements Runnable{
        /**
         * 全局事务id
         */
        private Long trxId;
        private long delay;
        private int retryTime;

        public RetryTrxTaskRunner(Long trxId,long delay,int retryTime){
            this.trxId = trxId;
            this.delay = delay;
            this.retryTime = retryTime;
        }

        @Override
        public void run() {
            //todo:重试执行任务, 执行成功：从runningSet移除，并调用updateTrxTaskSuccess
//            boolean locked = false;
//            String key = CacheUtils.getCompleteCacheKey("retryTrx", String.valueOf(trxId));
//            Lock lock = lockFactory.newLock(key,10*1000L);
//            try {
//                locked = lock.tryLock(1, TimeUnit.SECONDS);
//                log.info("get trx lock:" + key+","+locked);
//                if(locked){
//                    TrxTask trxTask = getTrxTaskByTrxId(trxId);
//                    if(trxTask == null){
//                        return;
//                    }
//                    Apps.getRequestContext().setTenantId(trxTask.getTenantId());
//                    if(!trxTask.getTaskStatus().equals(UserTrxConstants.TASK_RUNNING)){
//                        updateTrxTaskStatus(trxId,UserTrxConstants.TASK_RUNNING);
//                    }
//                    boolean b = TrxTaskUtil.trxInvoke(trxTask);
//                    if(b){
//                        //TODO: 待补充
//                        runningSet.remove(trxId);
//                        updateTrxTaskSuccess(trxId);
//                        return;
//                    }
//                    trxTask.setRetryTime(trxTask.getRetryTime()+1);
//                    trxTask.setTaskStatus(UserTrxConstants.TASK_FAILED);
//                    reliableUpdateTrxTask(trxTask,0,false);
//                }
//                if(retryTime>UserTrxConstants.TASK_RETRY_FAST_TIME){
//                    //如果执行超过重试上限，直接从运行队列移除，等待定时重试任务触发执行
//                    runningSet.remove(trxId);
//                    updateTrxTaskStatus(trxId,UserTrxConstants.TASK_FAILED);
//                    return;
//                }
//            } catch (Throwable e) {
//                log.error("trxTask retry error:"+trxId,e);
//                updateTrxTaskStatus(trxId,UserTrxConstants.TASK_FAILED);
//            }finally {
//                if(locked){
//                    lock.unlock();
//                }
//            }
            //如果执行失败，将自己放入后续执行队列
            long thisDelay = delay + 5L;
            RetryTrxTaskRunner retryTrxTaskRunner = new RetryTrxTaskRunner(trxId, thisDelay,retryTime+1);
            scheduledThreadPool.schedule(retryTrxTaskRunner,thisDelay,TimeUnit.SECONDS);
        }
    }
}
